#include <poll.h>
#include "threadPool.h"

typedef enum
{
	WORKER_CMD_STATUS_INVALID = (CMD_STATUS_LAST_VALUE + 1),
    WORKER_CMD_PING,                /*From server to worker*/
    WORKER_CMD_EXECUTE,             /*This cmd is sent from the server to worker*/
    WORKER_CMD_DONE,                /*This is sent from worker to server*/
    WORKER_CMD_EXIT,                /*Server to worker*/
	WORKER_CMD_STATUS_LAST_VALUE,
}WORKER_COMMAND;

typedef struct
{    
    pthread_t       thread_id;
    WORKER_COMMAND  cmd;

    /* For WORKER_CMD_PING the job_id should be 0, in all other cases
     * this should contain a valid job_id*/
    int             job_id;

    /*  arg1 will the return value in case of WORKER_CMD_DONE
     *  arg1 value will be returned back in case  of WORKER_CMD_PING
     *  arg1 should be zero in all other cases;
     */
    int             arg1;
    
    /*the below arguments are set in case of WORKER_CMD_EXECUTE*/
    int             (*function)(void *); 
    void            *fn_arg;
}WorkerMsg;


static void* worker_thread(void *arg);
static int processEventsFromClent(ThreadPool_MasterControl *t_mc);
static WorkerThreads* CreateWorkerThread(void);


static WorkerThreads* CreateWorkerThread(void)
{
    WorkerThreads *w = (WorkerThreads*)malloc(sizeof(WorkerThreads));
    if(NULL == w)
    {
        fprintf(stderr, "malloc(%d) failed\n", sizeof(WorkerThreads));
        return NULL;
    }
    if(0 != pipe(w->cmd_pipe_fd))
    {
        fprintf(stderr, "pipe() failed due to '%s'\n", strerror(errno));
        return NULL;
    }
    if(0 != pipe(w->response_pipe_fd))
    {
        close(w->cmd_pipe_fd[0]);
        close(w->cmd_pipe_fd[1]);
        fprintf(stderr, "pipe() failed due to '%s'\n", strerror(errno));
        return NULL;
    }
    w->current_job_id = 0;
    w->state = JOB_STATE_IDLE;
    if(0 != pthread_create(&w->thread_id, NULL, worker_thread, (void*)w))
    {
        close(w->cmd_pipe_fd[0]);
        close(w->cmd_pipe_fd[1]);
        close(w->response_pipe_fd[0]);
        close(w->response_pipe_fd[1]);
        w->thread_id = 0;
        return NULL;
    }
    return w;
}

void* worker_thread(void *arg)
{
    int e;
    WorkerMsg       msg;
    struct pollfd events[1];
    WorkerThreads   *w = (WorkerThreads*)arg;

    w->thread_id = pthread_self();

    printf("Started Worker : %#x\n", (unsigned int)w->thread_id);
    events[0].fd = w->cmd_pipe_fd[0];
    events[0].events = POLLIN | POLLPRI | POLLHUP;

    while(1)
    {
        events[0].revents = 0;
        e = poll(events, 1, -1);

        if(-1 == e)
        {
            fprintf(stderr, "poll() failed error : '%s' in thread %#x aborting\n", strerror(errno), (unsigned int)w->thread_id);
            abort();
        }
        else if(0 == e)
        {
            printf("Timeout... Continue\n");
            continue;
        }
        if((events[0].revents & POLLIN) || (events[0].revents & POLLPRI))
        {
            int count = read(w->cmd_pipe_fd[0], &msg, sizeof(WorkerMsg));
            if(count != sizeof(WorkerMsg))
            {
                fprintf(stderr, "Thread %#x, read() failed error : %s\n", (unsigned int)w->thread_id, strerror(errno));
                continue;
            }
            printf("\nWorker(%#x) CMD : %d\n",  (unsigned int)w->thread_id, msg.cmd);
            if(msg.thread_id != w->thread_id)
            {
                /* TODO: This message is not for Me.. but we use dedicated pipe right?
                 * how can this happen?, just continue for now*/

                continue;
            }
            switch(msg.cmd)
            {
                case WORKER_CMD_PING:
                       printf("WORKER_CMD_PING(%#x) received\n", (unsigned int)w->thread_id); 
                       write(w->response_pipe_fd[1], &msg, sizeof(WorkerMsg));
                       break;
                case WORKER_CMD_EXECUTE:
                      printf("WORKER_CMD_EXECUTE(%#x), job_id : %d starting--->\n", (unsigned int)w->thread_id, msg.job_id);
                      msg.arg1 = (*msg.function)(msg.fn_arg);
                      msg.cmd = WORKER_CMD_DONE;
                      write(w->response_pipe_fd[1], &msg, sizeof(WorkerMsg));
                      break;
                case WORKER_CMD_EXIT:
                      close(w->cmd_pipe_fd[0]);
                      close(w->cmd_pipe_fd[1]);
                      printf("Closing worker %#x\n", (unsigned int)w->thread_id);
                      pthread_exit(0);
                      break;
                default:                      
                      fprintf(stderr, "invalid cmd %d\n", msg.cmd);
                      break;
            }
        }
    }
    return 0;
}

void* threadpool_main_thread(void *arg)
{        
    struct pollfd events[1];
	int e, i;    
	ThreadPool_MasterControl *t_mc = (ThreadPool_MasterControl *)arg;

    t_mc->worker_list_size = t_mc->num_of_threads + 10;
    t_mc->worker_list = (WorkerThreads**)malloc(sizeof(WorkerThreads*) *  t_mc->worker_list_size);
    for(i = 0; i < t_mc->worker_list_size; i++)
    {
        if(i < t_mc->num_of_threads)
        {            
            t_mc->worker_list[i] = CreateWorkerThread();
        }
        else
        {
            t_mc->worker_list[i] = NULL;
        }
    }

    for(i = 0; i < t_mc->worker_list_size; i++)
    {
        if(t_mc->worker_list[i] == NULL)
                continue;
        WorkerMsg msg, msg1;
        msg.thread_id = t_mc->worker_list[i]->thread_id;
        msg.cmd = WORKER_CMD_PING;
        msg.arg1 = 0xffbbccaa;
        write(t_mc->worker_list[i]->cmd_pipe_fd[1], &msg, sizeof(WorkerMsg));
        read(t_mc->worker_list[i]->response_pipe_fd[0], &msg1, sizeof(WorkerMsg));
        if(msg.arg1 != msg1.arg1)
        {
            printf("ping failed to thread %#x\n", (unsigned int)msg.thread_id);
        }
        else
        {
            printf("pinged to thread %#x\n", (unsigned int)msg.thread_id);
        }
    }
    events[0].fd = t_mc->cmd_pipe_fd[0];
    events[0].events = POLLIN | POLLPRI | POLLHUP;

	printf("Thread Started, initial number of threads %d\n", t_mc->num_of_threads);
	while(1)
	{
        /*Reset the events*/
        events[0].revents = 0;
        e = poll(events, 1, -1);
        if(-1 == e)
        {
            fprintf(stderr, "poll() failed error : %s\n", strerror(errno));
            abort();
        }
        else if(0 == e)
        {
            fprintf(stderr, "timeout...\n");
            continue;
        }

        if((events[0].revents & POLLIN) || (events[0].revents & POLLPRI))
        {
            processEventsFromClent(t_mc);
        }
	}
	return NULL;
}

static int processEventsFromClent(ThreadPool_MasterControl *t_mc)
{
    int count;
    struct msg m;
    count = read(t_mc->cmd_pipe_fd[0], &m, sizeof(struct msg));
    if(count == -1)
    {
        fprintf(stderr, "read() failed aborting....\n");
        abort();
    }
    else if(count != sizeof(struct msg))
    {
        fprintf(stderr, "\nMessage not constructed properly size = %d, ignoreing this message\n", count);
        return 1;
    }

    switch(m.cmd)
    {
        case CMD_PING:
            {
                printf("CMD_PING received\n");
                /*Send back argument 1 */
                write(t_mc->response_pipe_fd[1], &m.arg1, sizeof(int));
                break;
            }
        case CMD_SET_THREAD_POOL_SIZE:
            {
                printf("CMD_SET_THREAD_POOL_SIZE to %d\n", m.arg1);
                t_mc->num_of_threads = m.arg1;
                break;
            }
        case CMD_GET_THREAD_POOL_SIZE:
            {
                printf("CMD_GET_THREAD_POOL_SIZE\n");
                write(t_mc->response_pipe_fd[1], &(t_mc->num_of_threads), sizeof(int));
                break;
            }
        case CMD_EXIT:
            {
                int done = 1, i;
                for(i = 0; i < t_mc->worker_list_size; i++)
                {
                    if(t_mc->worker_list[i] != NULL)
                    {
                        WorkerMsg msg;
                        msg.cmd = WORKER_CMD_EXIT;
                        msg.thread_id = t_mc->worker_list[i]->thread_id;
                        printf("closeing thread %#x\n", (unsigned int)msg.thread_id);
                        write(t_mc->worker_list[i]->cmd_pipe_fd[1], &msg, sizeof(WorkerMsg));
                        pthread_join(t_mc->worker_list[i]->thread_id, 0);
                    }
                }
                printf("Exiting server...\n");
                write(t_mc->response_pipe_fd[1], &done, sizeof(int));
                free(t_mc);
                pthread_exit(0);
            }
        default:
            printf("unknown command : %d\n", m.cmd);
            break;
    }

    return 0;
}
